﻿#include "RpcServer.h"
#include "XyNet/Channel/RpcBroadcastChannel.h"
#include "XyNet/Channel/RpcSocketChannel.h"
#include "XyNet/Channel/RpcSocketChannelManager.h"
#include "XyNet/Common/Cast.h"
#include "XyNet/Common/Logger.h"
#include "XyNet/Common/SetMapHelper.h"
#include "XyNet/Service/RpcServiceManager.h"

namespace XyNet
{
	RpcServer::Config::Config(const string& baseFileName) : RpcEndpoint::Config(baseFileName) {}

	//////////////////////////////////////////////////////////////////////////

	RpcServer::RpcServer(shared_ptr<Config> config)
		: RpcEndpoint("RpcServer", config)
		, _socketChannelManager(make_shared<RpcSocketChannelManager>())
	{
		Logger::getDefault()->log("Port: " + lexical_cast<string>(getConfig()->getPort()));
		_asioEndpoint = make_unique<boost::asio::ip::tcp::endpoint>(
			boost::asio::ip::tcp::v4(), numeric_cast<ushort>(getConfig()->getPort()));
		_acceptor = make_unique<boost::asio::ip::tcp::acceptor>(*getIoService(), *_asioEndpoint);
		//_acceptor->non_blocking(true);

		auto service = make_shared<ServerBaseServiceImpl>(this);
		getServiceManager()->addService(service);
		_baseService = service.get();

		for (uint i = 0; i < boost::thread::hardware_concurrency(); ++i)
			startAccept();
		run(boost::thread::hardware_concurrency() * 2);
		//run(1);
	}

	RpcServer::~RpcServer()
	{
		stop();
	}

	shared_ptr<const RpcServer::Config> RpcServer::getConfig() const
	{
		return XYNET_POINTER_DOWNCAST<const RpcServer::Config>(RpcEndpoint::getConfig());
	}

	shared_ptr<RpcBroadcastChannel> RpcServer::createBroadcastChannel()
	{
		return make_shared<RpcBroadcastChannel>(_socketChannelManager.get());
	}

	shared_ptr<RpcSocketChannel> RpcServer::findSocketChannel(const Id& id) const
	{
		return _socketChannelManager->findSocketChannel(id);
	}

	void RpcServer::handleSocketChannelException(
		shared_ptr<RpcSocketChannel> socketChannel, int errorCode)
	{
		RpcEndpoint::handleSocketChannelException(socketChannel, errorCode);

		function<void(weak_ptr<RpcSocketChannel>)> errorHandler =
			[this](weak_ptr<RpcSocketChannel> weakSocketChannel)
		{
			auto socketChannel = weakSocketChannel.lock();
			if (!socketChannel)
				return;

			_baseService->removeHeartBeatChecker(socketChannel->getId());
			_socketChannelManager->removeSocketChannel(socketChannel);

			// 记录连接数
			auto remoteEndpoint = socketChannel->getSocket()->remote_endpoint();
			boost::format format("%s:%d disconnected. Connect count %d");
			format % remoteEndpoint.address() % remoteEndpoint.port() %
				_socketChannelManager->getSocketChannelCount();
			Logger::getDefault()->log(format.str());
		};

		//if (errorCode != 0 && socketChannel->getSocket()->is_open())
		//{
		//	// 告诉客户端错误（如果可能的话）
		//	BaseService::SendErrorMethod method(socketChannel);
		//	method.getRequest()->set_error_code(errorCode);
		//	method.call([errorHandler, socketChannel]
		//	(BaseService::SendErrorMethod::Argument* argument)
		//	{
		//		errorHandler(socketChannel);
		//	}, [errorHandler, socketChannel]
		//	(BaseService::SendErrorMethod::Argument* argument)
		//	{
		//		errorHandler(socketChannel);
		//	});
		//}
		//else
		//{
		//	errorHandler(socketChannel);
		//}
		errorHandler(socketChannel);

		auto error = RpcErrors::createError(errorCode);
		Logger::getDefault()->logWarning(error->getInfo());
	}

	void RpcServer::startAccept()
	{
		auto socket = make_shared<boost::asio::ip::tcp::socket>(*getIoService());
		_acceptor->async_accept(*socket, bind(
			&RpcServer::handleAccept, this, socket, boost::asio::placeholders::error));
		//run();
	}

	void RpcServer::handleAccept(
		shared_ptr<boost::asio::ip::tcp::socket> socket, const boost::system::error_code& error)
	{
		if (error)
		{
			handleSocketException(socket, error);
			// 先前没有创建socketChannel，不需要handleSocketChannelException
		}
		else
		{
			//if (_socketChannelManager->getSocketChannelCount() >=
			//	getConfig()->getMaxConnection())
			//{
			//	// 超过最大连接数，断开
			//	socket->close();
			//}
			//else
			//{
			//	auto socketChannel = createSocketChannel(socket);
			//	_socketChannelManager->addSocketChannel(socketChannel);

			//	auto remoteEndpoint = socket->remote_endpoint();
			//	boost::format format("%s:%d accepted. Connect count %d");
			//	format % remoteEndpoint.address() % remoteEndpoint.port() %
			//		_socketChannelManager->getSocketChannelCount();
			//	Logger::getDefault()->log(format.str());

			//	socketChannel->start();
			//	_baseService->addHeartBeatChecker(socketChannel);	// 启动心跳检测
			//}

			auto socketChannel = createSocketChannel(socket);
			_socketChannelManager->addSocketChannel(socketChannel);

			// 记录连接数
			auto remoteEndpoint = socket->remote_endpoint();
			boost::format format("%s:%d accepted. Connect count %d");
			format % remoteEndpoint.address() % remoteEndpoint.port() %
				_socketChannelManager->getSocketChannelCount();
			Logger::getDefault()->log(format.str());

			_baseService->addHeartBeatChecker(socketChannel);	// 启动心跳检测
			socketChannel->start();

			if (_socketChannelManager->getSocketChannelCount() >
				getConfig()->getMaxConnection())
			{
				// 超过最大连接数，断开
				BaseService::SendErrorMethod method(socketChannel);
				method.getRequest()->set_error_code(RpcErrors::ReachMaxConnections);
				method.call([this, socketChannel](BaseService::SendErrorMethod::Argument* argument)
				{
					handleSocketChannelException(
						socketChannel, RpcErrors::ReachMaxConnections);
				});
			}
		}

		// 继续下一个接收
		if (!getIoService()->stopped())
			startAccept();
	}

	//////////////////////////////////////////////////////////////////////////

	ServerBaseServiceImpl::HeartBeatChecker::HeartBeatChecker(ServerBaseServiceImpl* baseService,
		shared_ptr<RpcSocketChannel> socketChannel)
		: _baseService(baseService)
		, _weakSocketChannel(socketChannel), _socketChannelId(socketChannel->getId())
		, _checkInterval(baseService->_checkInterval)
		, _timer(*baseService->_server->getIoService())
		, _hasHeartBeat(false)
	{
	}

	void ServerBaseServiceImpl::HeartBeatChecker::startTimer()
	{
		_timer.expires_from_now(_checkInterval);
		_timer.async_wait(bind(&HeartBeatChecker::handleCheckHeartBeat, shared_from_this(),
			boost::asio::placeholders::error));
	}

	void ServerBaseServiceImpl::HeartBeatChecker::sendHeartBeat()
	{
		_hasHeartBeat = true;
	}

	void ServerBaseServiceImpl::HeartBeatChecker::handleCheckHeartBeat(
		weak_ptr<HeartBeatChecker> weakChecker, const boost::system::error_code& error)
	{
		if (error)
			stopCheckerAndDisconnect(weakChecker, RpcErrors::SocketError);
		else
		{
			auto checker = weakChecker.lock();
			if (checker)
			{
				if (!checker->_hasHeartBeat)	// 没检测到心跳，断开
					stopCheckerAndDisconnect(weakChecker, RpcErrors::NotHeartBeat);
				else
				{
					auto socketChannel = checker->_weakSocketChannel.lock();
					if (!socketChannel)	// 通道失效，断开
						stopCheckerAndDisconnect(weakChecker, RpcErrors::NotConnected);
					else
					{
						checker->_hasHeartBeat = false;
						checker->startTimer();	// 开始下一次检测
					}
				}
			}
		}
	}

	void ServerBaseServiceImpl::HeartBeatChecker::stopCheckerAndDisconnect(weak_ptr<HeartBeatChecker> weakChecker, int errorCode)
	{
		auto checker = weakChecker.lock();
		if (checker)	// 除非服务端析构才无法Lock到
		{
			//// 删除心跳检测器
			//checker->_baseService->removeHeartBeatChecker(checker->_socketChannelId);

			// 删除套接字通道同时删除心跳检测器
			auto socketChannel = checker->_weakSocketChannel.lock();
			if (socketChannel)
			{
				checker->_baseService->_server->handleSocketChannelException(
					socketChannel, errorCode);
			}
		}
	}

	//////////////////////////////////////////////////////////////////////////

	ServerBaseServiceImpl::ServerBaseServiceImpl(RpcServer* server)
		: _server(server)
		, _checkInterval(boost::posix_time::seconds(
			_server->getConfig()->getCheckHeartBeatInterval()))
		//, _checkInterval(boost::posix_time::microsec(1))
	{
	}

	ServerBaseServiceImpl::~ServerBaseServiceImpl()
	{
		//cout << _idHeartBeatCheckerMapObject.createSharedLock()->getObject()->size() << endl;
	}

	void ServerBaseServiceImpl::onSendError(SendErrorMethod::Argument* argument)
	{
		auto errorCode = argument->getRequest()->error_code();
		auto error = argument->getController()->createError(errorCode);
		error->setDetail(argument->getRequest()->error_detail());
		Logger::getDefault()->logWarning(error->getInfo());
	}

	void ServerBaseServiceImpl::onSendHeartBeat(SendHeartBeatMethod::Argument* argument)
	{
		auto controller = argument->getController();
		auto lock = _idHeartBeatCheckerMapObject.createSharedLock();
		auto* idHeartBeatCheckerMap = lock->getObject();
		auto iter = idHeartBeatCheckerMap->find(controller->getSocketChannelId());
		if (iter != idHeartBeatCheckerMap->end())
		{
			lock->unlock();
			iter->second->sendHeartBeat();
		}
	}

	void ServerBaseServiceImpl::onGetUtcTime(GetUtcTimeMethod::Argument* argument)
	{
		throw std::logic_error("The method or operation is not implemented.");
	}

	void ServerBaseServiceImpl::onGetBrokerQueries(GetBrokerQueriesMethod::Argument* argument)
	{
		throw std::logic_error("The method or operation is not implemented.");
	}

	void ServerBaseServiceImpl::addHeartBeatChecker(shared_ptr<RpcSocketChannel> socketChannel)
	{
		auto checker = make_shared<HeartBeatChecker>(this, socketChannel);
		SetMapHelper::insert(_idHeartBeatCheckerMapObject.createUniqueLock()->getObject(),
			socketChannel->getId(), checker);
		checker->startTimer();
	}

	void ServerBaseServiceImpl::removeHeartBeatChecker(const Id& socketChannelId)
	{
		SetMapHelper::erase(_idHeartBeatCheckerMapObject.createUniqueLock()->getObject(),
			socketChannelId);
	}

}